//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// delete_executor.cpp
//
// Identification: src/execution/delete_executor.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

#include <algorithm>
#include <memory>
#include <vector>

#include "common/config.h"
#include "common/macros.h"
#include "common/rid.h"
#include "execution/executors/delete_executor.h"
#include "storage/table/tuple.h"
#include "type/value.h"
#include "type/value_factory.h"

namespace bustub {

DeleteExecutor::DeleteExecutor(ExecutorContext *exec_ctx, const DeletePlanNode *plan,
                               std::unique_ptr<AbstractExecutor> &&child_executor)
    : AbstractExecutor(exec_ctx),
      plan_(plan),
      child_executor_(std::move(child_executor)),
      table_info_(exec_ctx->GetCatalog()->GetTable(plan->table_oid_)) {}

void DeleteExecutor::Init() {
  child_executor_->Init();
  sum_ = 0;
}

auto DeleteExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  if (ending_) {
    return false;
  }
  std::vector<Value> values;
  values.reserve(plan_->OutputSchema().GetColumnCount());

  Tuple delete_tuple{};
  RID delete_rid{};

  auto all_index_info = GetExecutorContext()->GetCatalog()->GetTableIndexes(table_info_->name_);

  auto status = child_executor_->Next(&delete_tuple, &delete_rid);
  while (status) {
    BUSTUB_ENSURE(delete_rid.GetPageId() != INVALID_PAGE_ID, "invalid rid!");
    TupleMeta del_meta(table_info_->table_->GetTupleMeta(delete_rid));
    del_meta.is_deleted_ = true;
    table_info_->table_->UpdateTupleMeta(del_meta, delete_rid);

    for (auto index_info : all_index_info) {
      index_info->index_->DeleteEntry(
          delete_tuple.KeyFromTuple(table_info_->schema_, index_info->key_schema_, index_info->index_->GetKeyAttrs()),
          delete_rid, GetExecutorContext()->GetTransaction());
    }
    ++sum_;
    status = child_executor_->Next(&delete_tuple, &delete_rid);
  }

  values.push_back(ValueFactory::GetIntegerValue(sum_));
  Tuple output_tuple{values, &GetOutputSchema()};
  *tuple = output_tuple;

  ending_ = true;

  return true;
}

}  // namespace bustub
